package com.shure.queue.redisMqPs.config;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.shure.queue.redisMqPs.annotation.PsSubscribe;
import com.shure.queue.redisMqPs.subscribe.Subscribe;
import com.shure.queue.utils.ApplicationContextUtils;
import org.reflections.Reflections;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

@Configuration
public class SubscribeListenerConfig {
    /**
     * 初始化监听器
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);

        Map<String, Object> beans = ApplicationContextUtils.getBeansWithAnnotation(PsSubscribe.class);
        for (Map.Entry<String, Object> entry : beans.entrySet()) {
            Subscribe subscribe = (Subscribe) entry.getValue();

            System.out.println(subscribe.getClass().getName() + "listenerAdapter");
            MessageListenerAdapter listenerAdapterBean = ApplicationContextUtils.registerBean(subscribe.getClass().getName() + "listenerAdapter", MessageListenerAdapter.class, subscribe, "receiveMessage");

            PsSubscribe psSubscribe = AnnotationUtils.findAnnotation(subscribe.getClass(), PsSubscribe.class);
            container.addMessageListener(listenerAdapterBean, new PatternTopic(psSubscribe.key()));
        }

        //container.addMessageListener(listenerAdapter1, new PatternTopic("test"));
        //container.addMessageListener(listenerAdapter2, new PatternTopic("test"));

        /**
         * 如果不定义线程池，每一次消费都会创建一个线程，如果业务层面不做限制，就会导致秒杀超卖
         */
        ThreadFactory factory = new ThreadFactoryBuilder().setNamePrefix("redis-listener-pool-%d").build();
        Executor executor = new ThreadPoolExecutor(
                3,
                10,
                10L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                factory);
        container.setTaskExecutor(executor);
        return container;
    }
}
